package com.xian.scala.batch

import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.api.scala.{AggregateDataSet, DataSet, ExecutionEnvironment, GroupedDataSet}
import org.apache.flink.util.StringUtils
// 其中flatMap 和Map 中  需要引入隐式转换
import org.apache.flink.api.scala.createTypeInformation

/**
 * 读取文本文件 计算wordcount
 */
object ReadText {
  def main(args: Array[String]): Unit = {
    val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment

    val parameterTool: ParameterTool = ParameterTool.fromArgs(args)
    //传入参数  --infile e:/test/wc.txt
    var infile: String = parameterTool.get("infile")
    //用flink工具类 判断字符串参数为空
    if(StringUtils.isNullOrWhitespaceOnly(infile)){
      infile="e:/test/a.txt"
    }

    println("infile=="+infile)

    val lines: DataSet[String] = env.readTextFile(infile)
    lines.print()


    val words: DataSet[String] = lines.flatMap(line => line.split(" "))

    val word_one: DataSet[(String, Int)] = words.map(word => (word, 1))

    val word_group: GroupedDataSet[(String, Int)] = word_one.groupBy(0)

    val word_sum: AggregateDataSet[(String, Int)] = word_group.sum(1)


    word_sum.print()




  }

}
